<!DOCTYPE html>
<html lang="zh-cn">
<head>
    
    <link type="text/css" rel="stylesheet" href="/bundles/blog-common.css?v=KOZafwuaDasEedEenI5aTy8aXH0epbm6VUJ0v3vsT_Q1"/>
<link id="MainCss" type="text/css" rel="stylesheet" href="/skins/ThinkInside/bundle-ThinkInside.css?v=RRjf6pEarGnbXZ86qxNycPfQivwSKWRa4heYLB15rVE1"/>
<link type="text/css" rel="stylesheet" href="/blog/customcss/428549.css?v=%2fam3bBTkW5NBWhBE%2fD0lcyJv5UM%3d"/>

</head>
<body>
<a name="top"></a>

<div id="page_begin_html"></div><script>load_page_begin_html();</script>

<div id="topics">
	<div class = "post">
		<h1 class = "postTitle">
			<a id="cb_post_title_url" class="postTitle2" href="https://www.cnblogs.com/frankdeng/p/9425000.html">Kafka（五）Kafka的API操作和拦截器</a>
		</h1>
		<div class="clear"></div>
		<div class="postBody">
			<div id="cnblogs_post_body" class="blogpost-body"><h2>一 kafka的API操作</h2>
<h2 id="blogTitle25"><strong>1</strong><strong>.</strong><strong>1&nbsp;</strong><strong>环境准备</strong></h2>
<p>1）在eclipse中创建一个java工程</p>
<p>2）在工程的根目录创建一个lib文件夹</p>
<p>3）解压kafka安装包，将安装包libs目录下的jar包拷贝到工程的lib目录下，并build path。</p>
<p>4）启动zk和kafka集群，在kafka集群中打开一个消费者</p>
<p>[root@node21 kafka]$ bin/kafka-console-consumer.sh --zookeeper&nbsp;node21:2181,node22:2181,node23:2181&nbsp; --topic firstTopic</p>
<p>这里用maven，pom文件引入依赖</p>
<div class="cnblogs_code">
<pre>&lt;!-- https:<span style="color: #008000;">//</span><span style="color: #008000;">mvnrepository.com/artifact/org.apache.kafka/kafka --&gt;</span>
 &lt;dependency&gt;
    &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
    &lt;artifactId&gt;kafka_2.<span style="color: #800080;">11</span>&lt;/artifactId&gt;
     &lt;version&gt;<span style="color: #800080;">1.1</span>.<span style="color: #800080;">0</span>&lt;/version&gt;
 &lt;/dependency&gt;</pre>
</div>
<h2 id="blogTitle26"><strong>1</strong><strong>.</strong><strong>2 </strong><strong>生产者Java API</strong></h2>
<h3 id="blogTitle27"><strong>1.2.1</strong><strong>创建生产</strong><strong>者</strong><strong>（</strong><strong>过时</strong><strong>API）</strong></h3>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.producer;

import kafka.javaapi.producer.<span style="color: #ff0000;">Producer</span>;
import kafka.producer.<span style="color: #ff0000;">KeyedMessage</span>;
import kafka.producer.<span style="color: #ff0000;">ProducerConfig</span>;
import java.util.Properties;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> OldProducer {
    @SuppressWarnings(</span><span style="color: #800000;">"</span><span style="color: #800000;">deprecation</span><span style="color: #800000;">"</span><span style="color: #000000;">)
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) {

        Properties properties </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">metadata.broker.list</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">node21:9092,node22:9092,node23:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">request.required.acks</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">1</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">serializer.class</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">kafka.serializer.StringEncoder</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        Producer</span>&lt;Integer, String&gt; producer = <span style="color: #0000ff;">new</span> Producer&lt;Integer,String&gt;(<span style="color: #0000ff;">new</span><span style="color: #000000;"> ProducerConfig(properties));
        KeyedMessage</span>&lt;Integer, String&gt; message = <span style="color: #0000ff;">new</span> KeyedMessage&lt;Integer, String&gt;(<span style="color: #800000;">"</span><span style="color: #800000;">firstTopoic</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">hello world</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        producer.send(message );
    }
}</span></pre>
</div>
<h3 id="blogTitle28"><strong>1.2</strong><strong>.2 创建</strong><strong>生产者</strong><strong>（新</strong><strong>API</strong><strong>）</strong></h3>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.producer;
</span><span style="color: #000000;">
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> NewProducer {

    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) {

        Properties props </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> Kafka服务端的主机名和端口号</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">node21:9092,node22:9092,node23:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 等待所有副本节点的应答</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">acks</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">all</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 消息发送最大尝试次数</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">retries</span><span style="color: #800000;">"</span>, <span style="color: #800080;">0</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 一批消息处理大小</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">batch.size</span><span style="color: #800000;">"</span>, <span style="color: #800080;">16384</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 请求延时</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">linger.ms</span><span style="color: #800000;">"</span>, <span style="color: #800080;">1</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 发送缓存区内存大小</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">buffer.memory</span><span style="color: #800000;">"</span>, <span style="color: #800080;">33554432</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> key序列化</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">key.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> value序列化</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">value.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        KafkaProducer</span>&lt;String, String&gt; producer = <span style="color: #0000ff;">new</span> KafkaProducer&lt;&gt;<span style="color: #000000;">(props);
        </span><span style="color: #0000ff;">for</span> (<span style="color: #0000ff;">int</span> i = <span style="color: #800080;">0</span>; i &lt; <span style="color: #800080;">50</span>; i++<span style="color: #000000;">) {
            ProducerRecord</span>&lt;String, String&gt; record = <span style="color: #0000ff;">new</span> ProducerRecord&lt;String, String&gt;(<span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span>, Integer.toString(i), <span style="color: #800000;">"</span><span style="color: #800000;">hello world-</span><span style="color: #800000;">"</span> +<span style="color: #000000;">i);
            producer.send(record);
            System.</span><span style="color: #0000ff;">out</span><span style="color: #000000;">.println(record);
        }
        producer.close();
    }
}</span></pre>
</div>
<p><strong style="font-size: 1.17em;">1.2</strong><strong style="font-size: 1.17em;">.</strong><strong style="font-size: 1.17em;">3</strong><strong style="font-size: 1.17em;">&nbsp;创建</strong><strong style="font-size: 1.17em;">生产者</strong><strong style="font-size: 1.17em;">带</strong><strong style="font-size: 1.17em;">回调函数</strong><strong style="font-size: 1.17em;">（新</strong><strong style="font-size: 1.17em;">API</strong><strong style="font-size: 1.17em;">）</strong>&nbsp;</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.producer;

import java.util.Properties;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> CallBackProducer {

    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) {
        Properties props </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> Kafka服务端的主机名和端口号</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">node22:9092,node22:9092,node23:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 等待所有副本节点的应答</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">acks</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">all</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 消息发送最大尝试次数</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">retries</span><span style="color: #800000;">"</span>, <span style="color: #800080;">0</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 一批消息处理大小</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">batch.size</span><span style="color: #800000;">"</span>, <span style="color: #800080;">16384</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 增加服务端请求延时</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">linger.ms</span><span style="color: #800000;">"</span>, <span style="color: #800080;">1</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 发送缓存区内存大小</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">buffer.memory</span><span style="color: #800000;">"</span>, <span style="color: #800080;">33554432</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> key序列化</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">key.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> value序列化</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">value.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        KafkaProducer</span>&lt;String, String&gt; kafkaProducer = <span style="color: #0000ff;">new</span> KafkaProducer&lt;&gt;<span style="color: #000000;">(props);

        </span><span style="color: #0000ff;">for</span> (<span style="color: #0000ff;">int</span> i = <span style="color: #800080;">0</span>; i &lt; <span style="color: #800080;">10</span>; i++<span style="color: #000000;">) {
            kafkaProducer.send(</span><span style="color: #0000ff;">new</span> ProducerRecord&lt;String, String&gt;(<span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">hello</span><span style="color: #800000;">"</span> + i), <span style="color: #0000ff;">new</span><span style="color: #000000;"> Callback() {
                @Override
                </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> onCompletion(RecordMetadata metadata, Exception exception) {
                    </span><span style="color: #0000ff;">if</span> (metadata != <span style="color: #0000ff;">null</span><span style="color: #000000;">) {
                        System.</span><span style="color: #0000ff;">out</span>.println(metadata.partition() + <span style="color: #800000;">"</span><span style="color: #800000;">---</span><span style="color: #800000;">"</span> +<span style="color: #000000;"> metadata.offset());
                    }
                }
            });
        }
        kafkaProducer.close();
    }
}</span></pre>
</div>
<p>控制台打印输出如下：</p>
<div class="cnblogs_code">
<pre><span style="color: #800080;">1</span>---<span style="color: #800080;">17</span>
<span style="color: #800080;">1</span>---<span style="color: #800080;">18</span>
<span style="color: #800080;">1</span>---<span style="color: #800080;">19</span>
<span style="color: #800080;">1</span>---<span style="color: #800080;">20</span>
<span style="color: #800080;">2</span>---<span style="color: #800080;">11</span>
<span style="color: #800080;">2</span>---<span style="color: #800080;">12</span>
<span style="color: #800080;">2</span>---<span style="color: #800080;">13</span>
<span style="color: #800080;">0</span>---<span style="color: #800080;">22</span>
<span style="color: #800080;">0</span>---<span style="color: #800080;">23</span>
<span style="color: #800080;">0</span>---<span style="color: #800080;">24</span><span style="color: #000000;">

Process finished with exit code </span><span style="color: #800080;">0</span></pre>
</div>
<p><strong style="font-size: 1.17em;">1.2</strong><strong style="font-size: 1.17em;">.</strong><strong style="font-size: 1.17em;">4</strong><strong style="font-size: 1.17em;">&nbsp;自定义分区生产者</strong>&nbsp;</p>
<p>0）需求：将所有数据存储到topic的第0号分区上</p>
<p>1）定义一个类实现Partitioner接口，重写里面的方法（<span style="color: #ff0000;">过时API</span>）</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.producer;

import kafka.producer.<span style="color: #ff0000;">Partitioner</span>;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> OldCustomPartitioner implements Partitioner {

    </span><span style="color: #0000ff;">public</span><span style="color: #000000;"> OldCustomPartitioner() {
        super();
    }

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">int</span> partition(Object key, <span style="color: #0000ff;">int</span><span style="color: #000000;"> numPartitions) {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 控制分区</span>
        <span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;
    }
}</span></pre>
</div>
<p>2）自定义分区（<span style="color: #ff0000;">新API</span>）</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.producer;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> NewCustomPartitioner implements Partitioner {

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span> configure(Map&lt;String, ?&gt;<span style="color: #000000;"> configs) {
    }
    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">int</span> partition(String topic, Object key, <span style="color: #0000ff;">byte</span>[] keyBytes, Object value, <span style="color: #0000ff;">byte</span><span style="color: #000000;">[] valueBytes, Cluster cluster) {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 控制分区</span>
        <span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;
    }
    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> close() {
    }
}</span></pre>
</div>
<p>3）在代码中调用</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> PartitionerProducer {

    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) {

        Properties props </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> Kafka服务端的主机名和端口号</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">node21:9092,node22:9092,node23:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 等待所有副本节点的应答</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">acks</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">all</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 消息发送最大尝试次数</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">retries</span><span style="color: #800000;">"</span>, <span style="color: #800080;">0</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 一批消息处理大小</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">batch.size</span><span style="color: #800000;">"</span>, <span style="color: #800080;">16384</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 增加服务端请求延时</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">linger.ms</span><span style="color: #800000;">"</span>, <span style="color: #800080;">1</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 发送缓存区内存大小</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">buffer.memory</span><span style="color: #800000;">"</span>, <span style="color: #800080;">33554432</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> key序列化</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">key.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> value序列化</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">value.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 自定义分区</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">partitioner.class</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">com.xyg.kafka.NewCustomPartitioner</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        Producer</span>&lt;String, String&gt; producer = <span style="color: #0000ff;">new</span> KafkaProducer&lt;&gt;<span style="color: #000000;">(props);
        producer.send(</span><span style="color: #0000ff;">new</span> ProducerRecord&lt;String, String&gt;(<span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">1</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">kafka</span><span style="color: #800000;">"</span><span style="color: #000000;">));
        System.</span><span style="color: #0000ff;">out</span>.println(<span style="color: #0000ff;">new</span> ProducerRecord&lt;String, String&gt;(<span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">1</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">kafka</span><span style="color: #800000;">"</span><span style="color: #000000;">));
        producer.close();
    }
}</span></pre>
</div>
<p>（1）在node21上监控/opt/module/kafka/logs/目录下firstTopic主题3个分区的log日志动态变化情况）测试</p>
<p>[admin@node21 firstTopic-0]$ tail -f 00000000000000000000.log&nbsp;</p>
<p>[admin@node21 firstTopic-1]$ tail -f 00000000000000000000.log&nbsp;</p>
<p>[admin@node21 firstTopic-2]$ tail -f 00000000000000000000.log&nbsp;</p>
<p>（2）发现数据都存储到指定的分区了。</p>
<h2 id="blogTitle31"><strong>1</strong><strong>.</strong><strong>3 </strong><strong>消费者Java API</strong></h2>
<p>0）在控制台创建发送者</p>
<p>[root@node21 kafka]$ bin/kafka-console-producer.sh --broker-list node21:9092,node22:9092,node23:9092&nbsp; --topic firstTopic</p>
<p>&gt;hello world</p>
<h3><span style="font-size: 15px;">1.3.1</span><span style="font-size: 15px;"> 创建消费者（过时API）</span><strong style="font-size: 1.17em;"><br /></strong></h3>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.consume;</span><span style="color: #000000;">

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> CustomOldConsumer {

    @SuppressWarnings(</span><span style="color: #800000;">"</span><span style="color: #800000;">deprecation</span><span style="color: #800000;">"</span><span style="color: #000000;">)
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) {

        Properties properties </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">zookeeper.connect</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">node21:2181,node22:2181,node23:2181</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">group.id</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">g1</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">zookeeper.session.timeout.ms</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">500</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">zookeeper.sync.time.ms</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">250</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        properties.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">auto.commit.interval.ms</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">1000</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 创建消费者连接器</span>
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(<span style="color: #0000ff;">new</span><span style="color: #000000;"> ConsumerConfig(properties));
        HashMap</span>&lt;String, Integer&gt; topicCount = <span style="color: #0000ff;">new</span> HashMap&lt;&gt;<span style="color: #000000;">();
        topicCount.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span>, <span style="color: #800080;">1</span><span style="color: #000000;">);
        Map</span>&lt;String, List&lt;KafkaStream&lt;<span style="color: #0000ff;">byte</span>[], <span style="color: #0000ff;">byte</span>[]&gt;&gt;&gt; consumerMap =<span style="color: #000000;"> consumer.createMessageStreams(topicCount);
        KafkaStream</span>&lt;<span style="color: #0000ff;">byte</span>[], <span style="color: #0000ff;">byte</span>[]&gt; stream = consumerMap.<span style="color: #0000ff;">get</span>(<span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span>).<span style="color: #0000ff;">get</span>(<span style="color: #800080;">0</span><span style="color: #000000;">);
        ConsumerIterator</span>&lt;<span style="color: #0000ff;">byte</span>[], <span style="color: #0000ff;">byte</span>[]&gt; it =<span style="color: #000000;"> stream.iterator();

        </span><span style="color: #0000ff;">while</span><span style="color: #000000;"> (it.hasNext()) {
            System.</span><span style="color: #0000ff;">out</span>.println(<span style="color: #0000ff;">new</span><span style="color: #000000;"> String(it.next().message()));
        }
    }
}</span></pre>
</div>
<h3>1.3.2 创建消费者（<span style="color: #ff0000;">新API）</span></h3>
<p>官方提供案例（自动维护消费情况）</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">ackage com.xyg.kafka.consume;
</span><span style="color: #000000;">
import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> CustomNewConsumer {

    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) {

        Properties props </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 定义kakfa 服务的地址，不需要将所有broker指定上</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">node21:9092,node22:9092,node23:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 制定consumer group</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">group.id</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">test1</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 是否自动确认offset</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">enable.auto.commit</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">true</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 自动确认offset的时间间隔</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">auto.commit.interval.ms</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">1000</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> key的序列化类</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">key.deserializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringDeserializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> value的序列化类</span>
        props.put(<span style="color: #800000;">"</span><span style="color: #800000;">value.deserializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringDeserializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 定义consumer</span>
        KafkaConsumer&lt;String, String&gt; consumer = <span style="color: #0000ff;">new</span> KafkaConsumer&lt;&gt;<span style="color: #000000;">(props);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 消费者订阅的topic, 可同时订阅多个</span>
        consumer.subscribe(Arrays.asList(<span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">second</span><span style="color: #800000;">"</span>,<span style="color: #800000;">"</span><span style="color: #800000;">third</span><span style="color: #800000;">"</span><span style="color: #000000;">));
        </span><span style="color: #0000ff;">while</span> (<span style="color: #0000ff;">true</span><span style="color: #000000;">) {
            </span><span style="color: #008000;">//</span><span style="color: #008000;"> 读取数据，读取超时时间为100ms</span>
            ConsumerRecords&lt;String, String&gt; records = consumer.poll(<span style="color: #800080;">100</span><span style="color: #000000;">);
            </span><span style="color: #0000ff;">for</span> (ConsumerRecord&lt;String, String&gt;<span style="color: #000000;"> record : records)
                System.</span><span style="color: #0000ff;">out</span>.printf(<span style="color: #800000;">"</span><span style="color: #800000;">offset = %d, key = %s, value = %s%n</span><span style="color: #800000;">"</span><span style="color: #000000;">, record.offset(), record.key(), record.value());
        }
    }
}</span></pre>
</div>
<h2>二&nbsp;<strong>Kafka producer拦截器</strong></h2>
<h3 id="blogTitle33"><strong>2</strong><strong>.</strong><strong>1&nbsp;</strong><strong>拦截器原理</strong></h3>
<p>Producer拦截器(interceptor)是在Kafka 0.10版本被引入的，主要用于实现clients端的定制化控制逻辑。</p>
<p>对于producer而言，interceptor使得用户在消息发送前以及producer回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，producer允许用户指定多个interceptor按序作用于同一条消息从而形成一个拦截链(interceptor chain)。Intercetpor的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法包括：</p>
<p>（1）configure(configs)</p>
<p>获取配置信息和初始化数据时调用。</p>
<p>（2）onSend(ProducerRecord)：</p>
<p>该方法封装进KafkaProducer.send方法中，即它运行在用户主线程中。Producer确保在消息被序列化以及计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响目标分区的计算</p>
<p>（3）onAcknowledgement(RecordMetadata, Exception)：</p>
<p>该方法会在消息被应答或消息发送失败时调用，并且通常都是在producer回调逻辑触发之前。onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢producer的消息发送效率</p>
<p>（4）close：</p>
<p>关闭interceptor，主要用于执行一些资源清理工作</p>
<p>如前所述，interceptor可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个interceptor，则producer将按照指定顺序调用它们，并仅仅是捕获每个interceptor可能抛出的异常记录到错误日志中而非在向上传递。这在使用过程中要特别留意。</p>
<h3 id="blogTitle34"><strong>2</strong><strong>.</strong><strong>2&nbsp;</strong><strong>拦截器案例</strong></h3>
<p>1）需求：</p>
<p>实现一个简单的双interceptor组成的拦截链。第一个interceptor会在消息发送前将时间戳信息加到消息value的最前部；第二个interceptor会在消息发送后更新成功发送消息数或失败发送消息数。</p>
<p>&nbsp;<img src="https://images2018.cnblogs.com/blog/1385722/201806/1385722-20180603183433555-555336177.png" alt="" /></p>
<p>2）案例实操</p>
<p>（1）增加时间戳拦截器</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;
</span>
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span> TimeInterceptor implements ProducerInterceptor&lt;String, String&gt;<span style="color: #000000;"> {

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span> configure(Map&lt;String, ?&gt;<span style="color: #000000;"> map) {

    }
    @Override
    </span><span style="color: #0000ff;">public</span> ProducerRecord&lt;String, String&gt; onSend(ProducerRecord&lt;String, String&gt;<span style="color: #000000;"> record) {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 创建一个新的record，把时间戳写入消息体的最前部</span>
        <span style="color: #0000ff;">return</span> <span style="color: #0000ff;">new</span><span style="color: #000000;"> ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(),
                System.currentTimeMillis() </span>+ <span style="color: #800000;">"</span><span style="color: #800000;">,</span><span style="color: #800000;">"</span> +<span style="color: #000000;"> record.value().toString());
    }

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
    }<br />
    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> close() {
    }
}</span></pre>
</div>
<p>（2）统计发送消息成功和发送失败消息数，并在producer关闭时打印这两个计数器</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span> CounterInterceptor implements ProducerInterceptor&lt;String, String&gt;<span style="color: #000000;"> {
    </span><span style="color: #0000ff;">private</span> <span style="color: #0000ff;">int</span> errorCounter = <span style="color: #800080;">0</span><span style="color: #000000;">;
    </span><span style="color: #0000ff;">private</span> <span style="color: #0000ff;">int</span> successCounter = <span style="color: #800080;">0</span><span style="color: #000000;">;

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span> configure(Map&lt;String, ?&gt;<span style="color: #000000;"> configs) {
    }
    @Override
    </span><span style="color: #0000ff;">public</span> ProducerRecord&lt;String, String&gt; onSend(ProducerRecord&lt;String, String&gt;<span style="color: #000000;"> record) {
        </span><span style="color: #0000ff;">return</span><span style="color: #000000;"> record;
    }
    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> onAcknowledgement(RecordMetadata metadata, Exception exception) {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 统计成功和失败的次数</span>
        <span style="color: #0000ff;">if</span> (exception == <span style="color: #0000ff;">null</span><span style="color: #000000;">) {
            successCounter</span>++<span style="color: #000000;">;
        } </span><span style="color: #0000ff;">else</span><span style="color: #000000;"> {
            errorCounter</span>++<span style="color: #000000;">;
        }
    }
    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> close() {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 保存结果</span>
        System.<span style="color: #0000ff;">out</span>.println(<span style="color: #800000;">"</span><span style="color: #800000;">Successful sent: </span><span style="color: #800000;">"</span> +<span style="color: #000000;"> successCounter);
        System.</span><span style="color: #0000ff;">out</span>.println(<span style="color: #800000;">"</span><span style="color: #800000;">Failed sent: </span><span style="color: #800000;">"</span> +<span style="color: #000000;"> errorCounter);
    }
}</span></pre>
</div>
<p>（3）producer主程序</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.interceptor;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
</span>
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> InterceptorProducer {
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) throws Exception {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 1 设置配置信息</span>
        Properties props = <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">bootstrap.servers</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">node21:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">acks</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">all</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">retries</span><span style="color: #800000;">"</span>, <span style="color: #800080;">0</span><span style="color: #000000;">);
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">batch.size</span><span style="color: #800000;">"</span>, <span style="color: #800080;">16384</span><span style="color: #000000;">);
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">linger.ms</span><span style="color: #800000;">"</span>, <span style="color: #800080;">1</span><span style="color: #000000;">);
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">buffer.memory</span><span style="color: #800000;">"</span>, <span style="color: #800080;">33554432</span><span style="color: #000000;">);
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">key.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        props.put(</span><span style="color: #800000;">"</span><span style="color: #800000;">value.serializer</span><span style="color: #800000;">"</span>, <span style="color: #800000;">"</span><span style="color: #800000;">org.apache.kafka.common.serialization.StringSerializer</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 2 构建拦截链</span>
        List&lt;String&gt; interceptors = <span style="color: #0000ff;">new</span> ArrayList&lt;&gt;<span style="color: #000000;">();
        interceptors.add(</span><span style="color: #800000;">"</span><span style="color: #800000;">com.xyg.kafka.interceptor.TimeInterceptor</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        interceptors.add(</span><span style="color: #800000;">"</span><span style="color: #800000;">com.xyg.kafka.interceptor.CounterInterceptor</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
        String topic </span>= <span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span><span style="color: #000000;">;
        Producer</span>&lt;String, String&gt; producer = <span style="color: #0000ff;">new</span> KafkaProducer&lt;&gt;<span style="color: #000000;">(props);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 3 发送消息</span>
        <span style="color: #0000ff;">for</span> (<span style="color: #0000ff;">int</span> i = <span style="color: #800080;">0</span>; i &lt; <span style="color: #800080;">10</span>; i++<span style="color: #000000;">) {
            ProducerRecord</span>&lt;String, String&gt; record = <span style="color: #0000ff;">new</span> ProducerRecord&lt;&gt;(topic, <span style="color: #800000;">"</span><span style="color: #800000;">message</span><span style="color: #800000;">"</span> +<span style="color: #000000;"> i);
            producer.send(record);
        }
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 4 一定要关闭producer，这样才会调用interceptor的close方法</span>
<span style="color: #000000;">        producer.close();
    }
}</span></pre>
</div>
<p>3）测试</p>
<p>（1）在kafka上启动消费者，然后运行客户端java程序。</p>
<p>[root@node21 kafka]$ bin/kafka-console-consumer.sh --zookeeper&nbsp;node21:2181,node22:2181,node23:2181&nbsp; --from-beginning --topic firstTopic</p>
<p>1533465083631,message0<br />1533465084092,message3<br />1533465084092,message6<br />1533465084093,message9<br />1533465148033,message1<br />1533465148043,message4<br />1533465148044,message7<br />1533465154264,message0<br />1533465154650,message3<br />1533465154651,message6<br />1533465154651,message9</p>
<p>（2）观察java平台控制台输出数据如下：</p>
<p>Successful sent: 10</p>
<p>Failed sent: 0</p>
<h2>三&nbsp;<strong>kafka&nbsp;</strong><strong>Streams</strong></h2>
<h3 id="blogTitle37"><strong>3.1&nbsp; Kafka Streams概述</strong></h3>
<p><span>Kafka Streams是一个客户端库，用于构建任务关键型实时应用程序和微服务，其中输入和/或输出数据存储在Kafka集群中。</span><span>Kafka Streams结合了在客户端编写和部署标准Java和Scala应用程序的简单性以及Kafka服务器端集群技术的优势，使这些应用程序具有高度可扩展性，弹性，容错性，分布式等等。</span></p>
<h3 id="blogTitle38"><strong>3.2&nbsp; Kafka Streams</strong><strong>特点</strong></h3>
<p>1）功能强大&nbsp;</p>
<p>高扩展性，弹性，容错&nbsp;</p>
<p>2）轻量级&nbsp;</p>
<p>无需专门的集群&nbsp;</p>
<p>一个库，而不是框架</p>
<p>3）完全集成&nbsp;</p>
<p>100%的Kafka 0.10.0版本兼容</p>
<p>易于集成到现有的应用程序&nbsp;</p>
<p>4）实时性</p>
<p>毫秒级延迟&nbsp;</p>
<p>并非微批处理&nbsp;</p>
<p>窗口允许乱序数据&nbsp;</p>
<p>允许迟到数据</p>
<h3 id="blogTitle39"><strong>3.3&nbsp; 几种</strong><strong>Stream对比</strong></h3>
<p class="p">当前已经有非常多的流式处理系统，最知名且应用最多的开源流式处理系统有Spark Streaming和Apache Storm。Apache Storm发展多年，应用广泛，提供记录级别的处理能力，当前也支持SQL on Stream。而Spark Streaming基于Apache Spark，可以非常方便与图计算，SQL处理等集成，功能强大，对于熟悉其它Spark应用开发的用户而言使用门槛低。另外，目前主流的Hadoop发行版，如Cloudera和Hortonworks，都集成了Apache Storm和Apache Spark，使得部署更容易。</p>
<p class="p">既然Apache Spark与Apache Storm拥用如此多的优势，那为何还需要Kafka Stream呢？主要有如下原因。</p>
<p class="p">第一，Spark和Storm都是流式处理框架，而Kafka Stream提供的是一个基于Kafka的流式处理类库。框架要求开发者按照特定的方式去开发逻辑部分，供框架调用。开发者很难了解框架的具体运行方式，从而使得调试成本高，并且使用受限。而Kafka Stream作为流式处理类库，直接提供具体的类给开发者调用，整个应用的运行方式主要由开发者控制，方便使用和调试。</p>
<p class="p"><img src="https://images2018.cnblogs.com/blog/1385722/201806/1385722-20180603183520384-1523081879.png" alt="" /></p>
<p class="p">第二，虽然Cloudera与Hortonworks方便了Storm和Spark的部署，但是这些框架的部署仍然相对复杂。而Kafka Stream作为类库，可以非常方便的嵌入应用程序中，它对应用的打包和部署基本没有任何要求。</p>
<p class="p">第三，就流式处理系统而言，基本都支持Kafka作为数据源。例如Storm具有专门的kafka-spout，而Spark也提供专门的spark-streaming-kafka模块。事实上，Kafka基本上是主流的流式处理系统的标准数据源。换言之，大部分流式系统中都已部署了Kafka，此时使用Kafka Stream的成本非常低。</p>
<p class="p">第四，使用Storm或Spark Streaming时，需要为框架本身的进程预留资源，如Storm的supervisor和Spark on YARN的node manager。即使对于应用实例而言，框架本身也会占用部分资源，如Spark Streaming需要为shuffle和storage预留内存。但是Kafka作为类库不占用系统资源。</p>
<p class="p">第五，由于Kafka本身提供数据持久化，因此Kafka Stream提供滚动部署和滚动升级以及重新计算的能力。</p>
<p class="p">第六，由于Kafka Consumer Rebalance机制，Kafka Stream可以在线动态调整并行度。</p>
<h3 id="blogTitle40"><strong>3</strong><strong>.4&nbsp;</strong><strong>Stream</strong><strong>数据清洗</strong><strong>案例</strong></h3>
<p>0）需求：</p>
<p>实时处理单词带有&rdquo;&gt;&gt;&gt;&rdquo;前缀的内容。例如输入&rdquo;111&gt;&gt;&gt;hadoop&rdquo;，最终处理成&ldquo;hadoop&rdquo;</p>
<p>1）需求分析：</p>
<p><img src="https://images2018.cnblogs.com/blog/1385722/201808/1385722-20180805194937521-655773135.png" alt="" /></p>
<p>2）案例实操</p>
<p>（1）创建一个工程，pom文件引入依赖</p>
<div class="cnblogs_code">
<pre>&lt;!-- https:<span style="color: #008000;">//</span><span style="color: #008000;">mvnrepository.com/artifact/org.apache.kafka/kafka-streams --&gt;</span>
&lt;dependency&gt;
    &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
    &lt;artifactId&gt;kafka-streams&lt;/artifactId&gt;
    &lt;version&gt;<span style="color: #800080;">1.1</span>.<span style="color: #800080;">0</span>&lt;/version&gt;
&lt;/dependency&gt;</pre>
</div>
<p>（2）创建主类(TopologyBuilder是过时的)</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.stream;

import java.util.Properties;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.TopologyBuilder;
</span>
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> StreamApplication {
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) {
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 定义输入的topic</span>
        String <span style="color: #0000ff;">from</span> = <span style="color: #800000;">"</span><span style="color: #800000;">firstTopic</span><span style="color: #800000;">"</span><span style="color: #000000;">;
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 定义输出的topic</span>
        String to = <span style="color: #800000;">"</span><span style="color: #800000;">secondTopic</span><span style="color: #800000;">"</span><span style="color: #000000;">;
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 设置参数</span>
        Properties props = <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, </span><span style="color: #800000;">"</span><span style="color: #800000;">logFilter</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, </span><span style="color: #800000;">"</span><span style="color: #800000;">node21:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        StreamsConfig config </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> StreamsConfig(props);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 构建拓扑</span>
        <span style="color: #ff0000;">TopologyBuilder</span> builder = <span style="color: #0000ff;">new</span><span style="color: #000000;"> TopologyBuilder();
        builder.addSource(</span><span style="color: #800000;">"</span><span style="color: #800000;">SOURCE</span><span style="color: #800000;">"</span>, <span style="color: #0000ff;">from</span><span style="color: #000000;">)
                .addProcessor(</span><span style="color: #800000;">"</span><span style="color: #800000;">PROCESS</span><span style="color: #800000;">"</span>, <span style="color: #0000ff;">new</span> ProcessorSupplier&lt;<span style="color: #0000ff;">byte</span>[], <span style="color: #0000ff;">byte</span>[]&gt;<span style="color: #000000;">() {
                    @Override
                    </span><span style="color: #0000ff;">public</span> Processor&lt;<span style="color: #0000ff;">byte</span>[], <span style="color: #0000ff;">byte</span>[]&gt; <span style="color: #0000ff;">get</span><span style="color: #000000;">() {
                        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 具体分析处理</span>
                        <span style="color: #0000ff;">return</span> <span style="color: #0000ff;">new</span><span style="color: #000000;"> LogProcessor();
                    }
                }, </span><span style="color: #800000;">"</span><span style="color: #800000;">SOURCE</span><span style="color: #800000;">"</span><span style="color: #000000;">)
                .addSink(</span><span style="color: #800000;">"</span><span style="color: #800000;">SINK</span><span style="color: #800000;">"</span>, to, <span style="color: #800000;">"</span><span style="color: #800000;">PROCESS</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 创建kafka stream</span>
        KafkaStreams streams = <span style="color: #0000ff;">new</span><span style="color: #000000;"> KafkaStreams(builder, config);
        streams.start();
    }
}</span></pre>
</div>
<p>（3）具体业务处理</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.stream;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span> LogProcessor implements Processor&lt;<span style="color: #0000ff;">byte</span>[], <span style="color: #0000ff;">byte</span>[]&gt;<span style="color: #000000;"> {

    </span><span style="color: #0000ff;">private</span><span style="color: #000000;"> ProcessorContext context;

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> init(ProcessorContext context) {
        </span><span style="color: #0000ff;">this</span>.context =<span style="color: #000000;"> context;
    }

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span> process(<span style="color: #0000ff;">byte</span>[] key, <span style="color: #0000ff;">byte</span><span style="color: #000000;">[] value) {
        String input </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> String(value);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 如果包含&ldquo;&gt;&gt;&gt;&rdquo;则只保留该标记后面的内容</span>
        <span style="color: #0000ff;">if</span> (input.contains(<span style="color: #800000;">"</span><span style="color: #800000;">&gt;&gt;&gt;</span><span style="color: #800000;">"</span><span style="color: #000000;">)) {
            input </span>= input.split(<span style="color: #800000;">"</span><span style="color: #800000;">&gt;&gt;&gt;</span><span style="color: #800000;">"</span>)[<span style="color: #800080;">1</span><span style="color: #000000;">].trim();
            </span><span style="color: #008000;">//</span><span style="color: #008000;"> 输出到下一个topic</span>
            context.forward(<span style="color: #800000;">"</span><span style="color: #800000;">logProcessor</span><span style="color: #800000;">"</span><span style="color: #000000;">.getBytes(), input.getBytes());
        }</span><span style="color: #0000ff;">else</span><span style="color: #000000;">{
            context.forward(</span><span style="color: #800000;">"</span><span style="color: #800000;">logProcessor</span><span style="color: #800000;">"</span><span style="color: #000000;">.getBytes(), input.getBytes());
        }
    }

    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span> punctuate(<span style="color: #0000ff;">long</span><span style="color: #000000;"> timestamp) {
    }
    
    @Override
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> close() {
    }
}</span></pre>
</div>
<p>（4）运行程序</p>
<p>（5）在node21上启动生产者</p>
<p>[root@node21 kafka]$ bin/kafka-console-producer.sh --broker-list node21:9092,node22:9092,node23:9092&nbsp; --topic firstTopic</p>
<p>&gt;111&gt;&gt;&gt;hadoop<br />&gt;222&gt;&gt;&gt;spark<br />&gt;spark</p>
<p>（6）在node22上启动消费者</p>
<p>[root@node22 kafka]$ bin/kafka-console-consumer.sh&nbsp; --bootstrap-server node21:9092,node22:9092,node23:9092&nbsp;&nbsp;--from-beginning --topic secondTopic</p>
<p>hadoop<br />spark<br />spark</p>
<h3>3.5&nbsp; Stream官方wc案例</h3>
<p>参考文档：<a href="http://kafka.apache.org/11/documentation/streams/" target="_blank">http://kafka.apache.org/11/documentation/streams/</a></p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.kafka.stream;
</span><span style="color: #000000;">
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueStore;
import java.util.Arrays;
import java.util.Properties;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> WordCountApplication {

    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(final String[] args) throws Exception {
        Properties config </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, </span><span style="color: #800000;">"</span><span style="color: #800000;">wordcount-application</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, </span><span style="color: #800000;">"</span><span style="color: #800000;">node21:9092</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> StreamsBuilder();
        KStream</span>&lt;String, String&gt; textLines = builder.stream(<span style="color: #800000;">"</span><span style="color: #800000;">TextLinesTopic</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        KTable</span>&lt;String, Long&gt; wordCounts =<span style="color: #000000;"> textLines
                .flatMapValues(textLine </span>-&gt; Arrays.asList(textLine.toLowerCase().split(<span style="color: #800000;">"</span><span style="color: #800000;">\\W+</span><span style="color: #800000;">"</span><span style="color: #000000;">)))
                .groupBy((key, word) </span>-&gt;<span style="color: #000000;"> word)
                .count(Materialized.</span>&lt;String, Long, KeyValueStore&lt;Bytes, <span style="color: #0000ff;">byte</span>[]&gt;&gt;<span style="color: #0000ff;">as</span>(<span style="color: #800000;">"</span><span style="color: #800000;">counts-store</span><span style="color: #800000;">"</span><span style="color: #000000;">));
        wordCounts.toStream().to(</span><span style="color: #800000;">"</span><span style="color: #800000;">WordsWithCountsTopic</span><span style="color: #800000;">"</span><span style="color: #000000;">, Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> KafkaStreams(builder.build(), config);
        streams.start();
    }

}</span></pre>
</div>
<p>&nbsp;</p></div>

</body>
</html>
